package com.share.disruptor.trade;

import com.lmax.disruptor.*;

import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;

/**
 * @desc:
 * @author:caifan
 * @date:2021/5/22
 */
public class EventProcessorMain {
    private final static int BUFFER_SIZE = 1024;
    private final static int THREAD_NUMBERS = 4;

    public static void main(String[] args) throws Exception {
        /*
         * createSingleProducer创建一个单生产者的RingBuffer，
         * 第一个参数叫EventFactory，从名字上理解就是"事件工厂"，其实它的职责就是产生数据填充RingBuffer的区块。
         * 第二个参数是RingBuffer的大小，它必须是2的指数倍 目的是为了将求模运算转为&运算提高效率
         * 第三个参数是RingBuffer的生产都在没有可用区块的时候(可能是消费者（或者说是事件处理器） 太慢了)的等待策略
         */
        final RingBuffer<Trade> ringBuffer=RingBuffer.createSingleProducer(new EventFactory<Trade>() {
            @Override
            public Trade newInstance() {
                return new Trade();
            }
        }, BUFFER_SIZE,new YieldingWaitStrategy());

        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_NUMBERS);
        //  创建SequenceBarrier
        SequenceBarrier sequenceBarrier = ringBuffer.newBarrier();

        BatchEventProcessor<Trade> tradeBatchEventProcessor = new BatchEventProcessor<>(ringBuffer, sequenceBarrier, new TradeHandler());

        ringBuffer.addGatingSequences(tradeBatchEventProcessor.getSequence());

        executorService.submit(tradeBatchEventProcessor);

        Future<?> future = executorService.submit(new Callable<Trade>() {
            @Override
            public Trade call() throws Exception {
                long seq;
                for (int i = 0; i < 100; i++) {
                    seq = ringBuffer.next();// 占位 ringBuffer一个可用区块
                    // 放入数据
                    ringBuffer.get(seq).setPrice(Math.random() * 9);
                    // 发布数据
                    ringBuffer.publish(seq);
                }

                return null;
            }
        });
        future.get();
        Thread.sleep(1000);

        tradeBatchEventProcessor.halt();// 优雅结束
        executorService.shutdown();

    }
}
